package org.event.core.listener;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.event.core.construct.KafkaMesType;
import org.event.core.construct.KafkaTopics;
import org.event.core.entity.InspectorIssue;
import org.event.core.entity.MesBody;
import org.event.core.entity.Notice;
import org.event.core.service.consumer.KafkaConsumerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;

/**
 * @author zhangjm
 * @date 2023年11月29日 11:46
 */
public class KafkaConsumerListener {

    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerListener.class);
    @Autowired
    private KafkaConsumerService consumerService;

    /**
     * 功能描述: 监听指定topic，多个使用，
     * groupId：分组id
     * topics：监听当前topic数组
     * topic：监听单个topic
     */
    @KafkaListener(groupId = "${group.id}", topics = "#{'${consumer.topics}'.split(',')}", containerFactory = "")
    public void listener(ConsumerRecord<String, String> consumerRecord) {
        logger.info("开始消费" + KafkaTopics.SD_JY_DUTY_TOPIC.getTopicName() + "的消息{}", consumerRecord.value());
        MesBody mesBody = JSONObject.parseObject(consumerRecord.value(), MesBody.class);
        logger.error("kafka监听-当前消息类型：" + mesBody.getCommand());
        //督查督办
        if (mesBody.getCommand().equals(KafkaMesType.SD_INSPECTOR_ISSUE.getMesCode())
                || mesBody.getCommand().equals(KafkaMesType.SD_INSPECT_DISPOSE.getMesCode())
                || mesBody.getCommand().equals(KafkaMesType.SD_INSPECT_RES.getMesCode())) {
            logger.error("督查督办监听消息处理开始----->----->");
            InspectorIssue inspectorIssue = JSONObject.parseObject(mesBody.getData(), InspectorIssue.class);
            consumerService.inspectorListener(inspectorIssue);
        }

        //通知通报
        if (mesBody.getCommand().equals(KafkaMesType.SD_NOTICE_ISSUE.getMesCode())) {
            logger.error("通知通报开始监听");
            Notice notice = JSONObject.parseObject(mesBody.getData(), Notice.class);
            consumerService.noticeListener(notice);
        }
    }


}
